package drds.binlog.parse.mysql.dbsync;

import drds.binlog.binlog_event.LogFetcher;
import drds.binlog.common.Authors;
import drds.binlog.database_driver.socket.SocketChannel;
import drds.common.Author;
import drds.common.Constants;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.SocketTimeoutException;
import java.nio.channels.ClosedByInterruptException;


@Slf4j
@Author(name = Authors.LI_YANG)
public class SocketLogFetcher extends LogFetcher
{

    // Master heartbeat interval
    public static final int master_heartbeat_period_seconds = 15;

    /**
     * Packet headerPacket sizes
     */
    public static final int net_header_size = 4;
    public static final int sqlstate_length = 5;
    /**
     * Packet offsets
     */
    public static final int packet_length_offset = 0;
    public static final int packet_sequence_offset = 3;
    /**
     * Maximum packet length
     */
    public static final int max_packet_length = (256 * 256 * 256 - 1);

    // +10s 确保 timeout > heartbeat interval
    private static final int read_timeout_milliseconds = (master_heartbeat_period_seconds + 10) * 1000;
    @Setter
    @Getter
    private SocketChannel socketChannel;
    @Setter
    @Getter
    private boolean semi = false;


    public SocketLogFetcher(final int initialCapacity)
    {
        super(initialCapacity, default_growth_factor);
    }


    public void start(SocketChannel socketChannel) throws IOException
    {
        this.socketChannel = socketChannel;
        String semi = System.getProperty("db.semi");
        if ("true".equals(semi))
        {
            this.semi = true;
        }
    }


    public boolean fetch() throws IOException
    {
        try
        {
            // fetch header packet
            if (!read(0, net_header_size))
            {
                log.warn("Reached end of input stream while fetching headerPacket");
                return false;
            }

            // Fetching the first packet(may a multi-packet).
            int netlength = getLittleEndian3UnsignedInt(packet_length_offset);
            int netnum = get1UnsignedInt(packet_sequence_offset);
            if (!read(net_header_size, netlength))
            {
                log.warn("Reached end of input stream: packet #" + netnum + ", len = " + netlength);
                return false;
            }

            // Detecting error code.
            final int mark = get1UnsignedInt(net_header_size);
            if (mark != 0)
            {
                if (mark == 255) // error from master
                {
                    // Indicates an error, for example trying to fetch from
                    // wrong
                    // binlog setIndexAsOriginIndexAndOffset$originIndex.
                    index = net_header_size + 1;
                    final int errorNum = getLittleEndian2SignedIntByIndex();
                    String sqlstate = forwardByIndex(1).getFixString(sqlstate_length);
                    String errorMessage = getFixString(limit - index);
                    throw new IOException("Received error packet:" + " errno = " + errorNum + ", sqlstate = " + sqlstate
                            + " errmsg = " + errorMessage);
                } else if (mark == 254)
                {
                    // Indicates end of stream. It's not removeEventExecuteDateTimeLongFromQueue createDateTimeLong this would
                    // be sent.
                    log.warn("Received EOF packet from server, apparent"
                            + " master disconnected. It's may be duplicateByIndex slaveId , check instance config");
                    return false;
                } else
                {
                    // Should not happen.
                    throw new IOException("Unexpected response " + mark + " while fetching binlog: packet #" + netnum + ", len = " + netlength);
                }
            }

            // if mysql is in semi mode
            if (semi)
            {
                // parse semi mark
                int semimark = get1UnsignedInt(net_header_size + 1);
                if (Constants.developMode)
                {
                    log.debug("semimark:" + semimark);
                }
                int semival = get1UnsignedInt(net_header_size + 2);
                this.semival = semival;
            }

            // The first packet is a multi-packet, concatenate the packets.
            while (netlength == max_packet_length)
            {
                if (!read(0, net_header_size))
                {
                    log.warn("Reached end of input stream while fetching headerPacket");
                    return false;
                }

                netlength = getLittleEndian3UnsignedInt(packet_length_offset);
                netnum = get1UnsignedInt(packet_sequence_offset);
                if (!read(limit, netlength))
                {
                    log.warn("Reached end of input stream: packet #" + netnum + ", len = " + netlength);
                    return false;
                }
            }

            // Preparing bytes variables to decoding.
            if (semi)
            {
                originIndex = net_header_size + 3;
            } else
            {
                originIndex = net_header_size + 1;
            }
            index = originIndex;
            limit -= originIndex;
            return true;
        } catch (SocketTimeoutException e)
        {
            close(); /* Do cleanup */
            log.error("Socket timeout expired, closing connection", e);
            throw e;
        } catch (InterruptedIOException e)
        {
            close(); /* Do cleanup */
            log.info("I/O interrupted while reading from client socket", e);
            throw e;
        } catch (ClosedByInterruptException e)
        {
            close(); /* Do cleanup */
            log.info("I/O interrupted while reading from client socket", e);
            throw e;
        } catch (IOException e)
        {
            close(); /* Do cleanup */
            log.error("I/O error while reading from client socket", e);
            throw e;
        }
    }

    private final boolean read(final int offset, final int length) throws IOException
    {
        ensureCapacity(offset + length);
        socketChannel.read(bytes, offset, length, read_timeout_milliseconds);
        if (limit < offset + length)
        {
            limit = offset + length;
        }
        return true;
    }


    public void close() throws IOException
    {
        // do nothing
    }

}
